# coding: utf-8

import time
import json
import happybase
from pandas import DataFrame
from concurrent.futures import ThreadPoolExecutor

all_columns = [
    b'actions:user_id',
    b'article:article_id',
    b'actions:action_id',
    b'actions:timestamp'
]


def handler_row_score(action_id, timestamp):
    # print(action_id, type(action_id))
    """ 处理数据分数
    :param row_dict:
        action_id
            1 刷文章列表
            2 查看文章
            3 评论文章
            4 收藏文章
            5 退出文章详情
            6 打开app
            7 隐藏app
            8 取消收藏
            9 展开摘要
           10 收起摘要
    :return: row_dict
    """
    current_time = time.time()
    action_score_dict = {
        # 1: -1.0,
        2: 5.0,
        3: 10.0,
        4: 20.0,
        # 8: -2.0,
        9: 2.0
    }
    if action_id not in action_score_dict:
        return
    time_weight = 0.9 ** ((current_time - timestamp) / 86400)
    return action_score_dict[action_id] * time_weight


def handler_row(data):
    # flag = False
    # for column in all_columns:
    #     if column not in data:
    #         flag = True
    #         break
    # if flag:
    #     return
    user_id = data[b'actions:user_id'].decode('utf-8')
    news_id = int(data[b'article:article_id'].decode('utf-8'))
    timestamp = int(data[b'actions:timestamp'].decode('utf-8'))
    action_id = int(data[b'actions:action_id'].decode('utf-8'))
    print('%s|%s|%s|%s' % (user_id, news_id, action_id, timestamp))
    if news_id == 0:
        return
    score = handler_row_score(action_id, timestamp)
    if not score:
        return
    return [user_id, news_id, action_id, timestamp, score]


def get_data_frame():
    current_last_row = '1000072324_1551675041_189246'
    current_last_timestamp = None
    connection = happybase.Connection('hb-uf6i3710q15t89giw-001.hbase.rds.aliyuncs.com', 9099)
    table = happybase.Table(b'news', connection)
    query = table.scan(row_start=current_last_row, limit=100000)
    data_list = []
    futures = []
    pool = ThreadPoolExecutor(1000)
    for (key, data) in query:
        futures.append(pool.submit(handler_row, data))
        current_last_row = key
        current_last_timestamp = int(data[b'actions:timestamp'].decode('utf-8'))
        # print(current_last_row)
    while futures:
        future = futures.pop()
        if not future.done():
            futures.append(future)
            continue
        results = future.result()
        if not results:
            continue
        data_list.append(results)
    pool.shutdown()
    print('-----------data_list: %s-------------' % len(data_list))
    columns = ['user_id', 'news_id', 'action_id', 'timestamp', 'score']
    print(current_last_row, time.localtime(current_last_timestamp))
    return DataFrame(data=data_list, columns=columns)


def main():
    data_frame = get_data_frame()
    print(data_frame)
    print('----------data_frame: %s---------' % data_frame.size)
    new_data_frame = data_frame.groupby(by=['user_id', 'news_id']).agg({'score': 'sum'})
    print('----------new_data_frame: %s---------' % new_data_frame.size)
    # print(json.dumps(dir(new_data_frame)))
    print(new_data_frame)
    for (key, row) in new_data_frame.iterrows():
        user_id, news_id = key
        score = row['score']
        print('user_id: %s, news_id: %s, score: %s' % (user_id, news_id, score))



if __name__ == '__main__':
    main()
